package com.iot.spi.core;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.util.StrUtil;
import com.iot.spi.config.IotSpiConfig;
import com.iot.spi.core.emqx.contextaware.ProductPackHandlerManagement;
import com.iot.spi.core.emqx.handler.EmqxProductPackHandler;
import com.iot.spi.exception.IotSpiException;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 控制端
 */
@Slf4j
@Component
@AutoConfigureAfter({IotSpiConfig.class})
@DependsOn({"ProductPackHandlerManagement"})
public class IotEmqxClient implements ApplicationListener<ApplicationStartedEvent> {

    @Resource
    private IotSpiConfig config;

    @Autowired
    private ProductPackHandlerManagement packHandlerManagement;

    private void checkConfig() {
        if (StrUtil.isBlank(config.getSpi().getHost())){
            throw new IotSpiException("iot host is null");
        }if (StrUtil.isBlank(config.getSpi().getAppKey())){
            throw new IotSpiException("iot appkey is null");
        }if (StrUtil.isBlank(config.getSpi().getAppSecret())){
            throw new IotSpiException("iot AppSecret is null");
        }if (StrUtil.isBlank(config.getEmqx().getHost())){
            throw new IotSpiException("iot emqx host is null");
        }if (StrUtil.isBlank(config.getEmqx().getUserName())){
            throw new IotSpiException("iot emqx userName is null");
        }if (StrUtil.isBlank(config.getEmqx().getPassword())){
            throw new IotSpiException("iot emqx password is null");
        }
    }

    private MqttClient client;

    private static ExecutorService service = Executors.newFixedThreadPool(6);

    @Override
    public void onApplicationEvent(ApplicationStartedEvent event) {
        if (config.getEnabled()){
            checkConfig();
            initEmqx();
        }
    }

    private void initEmqx() {
        try {
            client = new MqttClient(config.getEmqx().getHost(),config.getSpi().getAppKey()+"_"+System.currentTimeMillis(), new MemoryPersistence());
            client.setCallback(new Callback());
            while (!client.isConnected()) {
                client.connect(getOption(config));
                try {
                    TimeUnit.MILLISECONDS.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            String topic = config.getSpi().getAppKey()+"/#";
            client.subscribe(topic);
            log.info("emqx connect OK ... subscribe topic {}",topic);
        } catch (MqttException e) {
            log.error(" emqx 连接失败 ... {}", ExceptionUtil.stacktraceToString(e));
            throw new RuntimeException(e);
        }

    }


    /**
     * MQTT的连接设置
     *
     * @param emqProperties
     * @return
     */
    public MqttConnectOptions getOption(IotSpiConfig emqProperties) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(emqProperties.getEmqx().getUserName());
        options.setPassword(emqProperties.getEmqx().getPassword().toCharArray());
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(true);
        //断线重连
        options.setAutomaticReconnect(true);
        // 设置超时时间 单位为秒
        options.setConnectionTimeout(30);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*10秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        options.setKeepAliveInterval(10);
        return options;
    }

    class Callback implements MqttCallback {
        public void connectionLost(Throwable cause) {
            initEmqx();
        }

        public void messageArrived(String topic, MqttMessage message) throws Exception {
            String[] item = topic.split("/");
            String appKey = item[0];
            String productCode = item[1];
            String deviceModel = item[2];
            String deviceId = item[3];
            EmqxProductPackHandler res = packHandlerManagement.getPackHandler(productCode+"_"+deviceModel);
            service.submit(() -> res.handle(new String(message.getPayload())));
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {

        }
    }

}
